
package readImposeWriteConsult;

import java.util.HashSet;
import java.util.Set;
import java.util.Vector;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import assignment1.perfectfailuredetector.CrashEvent;
import assignment1.perfectfailuredetector.PerfectFailureDetector;
import atomicRegister.AtomicRegister;
import atomicRegister.ReadRequest;
import atomicRegister.ReadResponse;
import atomicRegister.WriteRequest;
import atomicRegister.WriteResponse;

import BEB.BebBroadcast;
import BEB.BebDeliver;
import BEB.BEB;

import se.kth.ict.id2203.pp2p.PerfectPointToPointLink;
import se.kth.ict.id2203.pp2p.Pp2pDeliver;
import se.kth.ict.id2203.pp2p.Pp2pSend;
import se.sics.kompics.ComponentDefinition;
import se.sics.kompics.Handler;
import se.sics.kompics.Kompics;
import se.sics.kompics.Negative;
import se.sics.kompics.Positive;
import se.sics.kompics.Start;
import se.sics.kompics.address.Address;

/**
 * The <code>SimpleUnreliableBroadcast</code> class.
 * 
 * @author Federico Piccinini <fpic@sics.se>
 */
public final class ReadImposeWriteConsult extends ComponentDefinition {

	Positive<BEB> beb =  requires(BEB.class);	
	Positive<PerfectPointToPointLink> pp2p = requires (PerfectPointToPointLink.class);
	Positive<PerfectFailureDetector> pfd = requires (PerfectFailureDetector.class);
	
	Negative<AtomicRegister> areg = provides (AtomicRegister.class);

	private static final Logger logger = LoggerFactory
			.getLogger(ReadImposeWriteConsult.class);

	private Set<Address> neighborSet;
	private Address self;
	private int maxR;
	private Set<Address> correct;
	private int mrankI;
	private Vector<Set<Address>> writeSet;
	private Vector<Boolean> reading;
	private Vector<Integer> reqId;
	private Vector<Integer> readVal;
	private Vector<Integer> v;
	private Vector<Integer> ts;
	private Vector<Integer> mrank;
	
	/**
	 * Instantiates a new application0.
	 */
	public ReadImposeWriteConsult() {
		subscribe(handleInit, control);
		subscribe(handleStart, control);
		subscribe(handleCrash, pfd);
		subscribe(handleReadRequest, areg);
		subscribe(handleWriteRequest, areg);
		subscribe(handleWriteMessage, beb);
		subscribe(ackMessageHandler, pp2p);
	}

	Handler<RiwcInit> handleInit = new Handler<RiwcInit>() {
		public void handle(RiwcInit event) {
			logger.info("Received Init event");
			neighborSet = event.getNeighborSet();
			self = event.getSelf();
			neighborSet.add(self);
			maxR = event.getMaxR();
			mrankI = self.getId();
			
			writeSet = new Vector<Set<Address>>(maxR); 
			reading = new Vector<Boolean>(maxR);
			reqId = new Vector<Integer> (maxR);
			readVal = new Vector<Integer> (maxR);
			v = new Vector<Integer> (maxR);
			ts = new Vector<Integer>(maxR);
			mrank = new Vector<Integer>(maxR);
			
			correct = new HashSet<Address>(neighborSet);
			
			for (int i = 0; i<maxR; ++i) {
				writeSet.add(i, new HashSet<Address>());
				reading.add(i, false);
				reqId.add(i, 0);
				readVal.add(i, 0);
				v.add(i, 0);
				ts.add(i, 0);
				mrank.add(i, 0);
			}
		}
	};

	Handler<Start> handleStart = new Handler<Start>() {
		public void handle(Start event) {
		}
	};

	
	Handler <CrashEvent> handleCrash = new Handler<CrashEvent>() {
		public void handle (CrashEvent event) {
			correct.remove(event.getSource());
		}
	};
	
	Handler <ReadRequest> handleReadRequest = new Handler<ReadRequest> () {
		public void handle (ReadRequest event) {
			logger.info( "ReadRequest on register {}", event.getR());
			int r = event.getR();
			reqId.set(r, (reqId.get(r)+1));
			reading.set(r, true);
			writeSet.get(r).clear();
			readVal.set(r, v.get(r));
			trigger (new BebBroadcast(new WriteMessage(self, r, reqId.get(r), ts.get(r), mrank.get(r), v.get(r))), beb);
		}
	};
	
	Handler <WriteRequest> handleWriteRequest = new Handler<WriteRequest> () {
		public void handle (WriteRequest event) {
			logger.info( "WriteRequest ({}, {})", event.getR(), event.getVal());
			int r = event.getR();
			reqId.set(r, (reqId.get(r)+1));
			writeSet.set(r, new HashSet<Address>());
			trigger (new BebBroadcast(new WriteMessage(self, r, reqId.get(r), ts.get(r)+1, mrankI, event.getVal())), beb);	
		}
	};
	
	Handler <WriteMessage> handleWriteMessage = new Handler<WriteMessage> () {
		public void handle (WriteMessage event) {
			int r = event.getR();
			if (timestampIsGreater(event.getTs(), event.getMrank(), ts.get(r), mrank.get(r))) {
				v.set(r, event.getValue());
				ts.set(r, event.getTs());
				mrank.set(r, event.getMrank());
			}
			trigger (new Pp2pSend (event.getSource(), new AckMessage(self, r, event.getReqid())), pp2p);
		}
	};
	
	
	Handler <AckMessage> ackMessageHandler = new Handler<AckMessage> () {
		public void handle (AckMessage event) {
			int r = event.getR();
			writeSet.get(r).add(event.getSource());
			onWriteSetChange (r);
		} 
	};
	
	private boolean timestampIsGreater (int ts1, int processId1, int ts2, int processId2) {
		if (ts1 > ts2 ) {
			return true;
		} else if (ts1 == ts2) {
			return (processId1>processId2);
		}
		return false;
	}
	
	private void onWriteSetChange (int r) {
		if (writeSet.get(r).containsAll(correct)) {
			if (reading.get(r)==true) {
				reading.set(r, false);
				trigger (new ReadResponse(r, readVal.get(r)), areg);
			} else {
				trigger (new WriteResponse(r), areg);
			}
		}
	}
	
	
	private void doShutdown() {
		System.out.println("2DIE");
		System.out.close();
		System.err.close();
		Kompics.shutdown();
	}
}
